/********************************************************/
/*	Author:		gong_libin			*/
/*	Date:		2017_11_01			*/
/*	File:		CmKafkaTest.cpp			*/
/********************************************************/

#include "CmUtil.h"
#include "CmKafka.h"

void CmKafkaTestCons()
{
	UINT uiCount = 0;
	CMSTRING_C* pCMsg = NULL;
	CmCls::CCmQueue* pCQueue = NULL;
	CCmConsumer CCons((char*)"192.168.0.7:9092", (char*)"t-ad-operation", (char*)"group_test");

	while (true) {
		if (NULL != (pCQueue = CCons.CmConsumerRecv())) {
			pCQueue->CmQueuePop(&pCMsg, sizeof(CMSTRING_C**));
			CM_PRINT("%04d: %s\n", ++ uiCount, pCMsg->c_str());
			delete pCMsg;
		}
	}

	return;
}

void CmKafkaTestProd()
{
	char szTime[CM_DAYTIME] = {'\0'};
	char szPacket[CM_BUFFER] = {'\0'};
	CCmProducer CProd((char*)"192.168.0.7:9092", (char*)"t-ad-operation");

	CmLocalDayTime(szTime);
	for (UINT uiIdx = 0; uiIdx < 5; uiIdx ++) {
		sprintf(szPacket, "%s + %d", szTime, uiIdx);
		CProd.CmProducerSend(CM_KFK_PART, szPacket, strlen(szPacket));
	}

	CProd.CmKafkaPoll(CM_KFK_TOUT);

	return;
}

int main(int argc, char* argv[])
{
	if (argc > 1) {
		CmKafkaTestCons();
	}
	else {
		CmKafkaTestProd();
	}

	return CM_SUCCESS;
}
